/**
 * 
 */
package org.demetra.jets.bdb;

import java.io.FileNotFoundException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.Iterator;
import java.util.SortedSet;

import org.demetra.jets.TupleSet;
import org.demetra.jets.TupleStoreException;

import com.ericsson.otp.erlang.OtpErlangBinary;
import com.ericsson.otp.erlang.OtpErlangObject;
import com.ericsson.otp.erlang.OtpErlangString;
import com.ericsson.otp.erlang.OtpErlangTuple;
import com.sleepycat.bind.EntryBinding;
import com.sleepycat.bind.serial.SerialBinding;
import com.sleepycat.bind.serial.StoredClassCatalog;
import com.sleepycat.collections.StoredMap;
import com.sleepycat.collections.StoredSortedMap;
import com.sleepycat.db.Database;
import com.sleepycat.db.DatabaseConfig;
import com.sleepycat.db.DatabaseEntry;
import com.sleepycat.db.DatabaseException;
import com.sleepycat.db.DatabaseType;
import com.sleepycat.db.Environment;
import com.sleepycat.db.SecondaryConfig;
import com.sleepycat.db.SecondaryDatabase;

/**
 * @author vjache
 *
 */
public class BdbOtpTupleSet implements
		TupleSet<OtpErlangTuple> {

	private static final String CLASS_CATALOG = "java_class_catalog";
	private static final String INDEX_PREFIX = "index_";
	
	private Database _db;
	private StoredMap _map;
	private Environment _env;
	private EntryBinding _keyBinding;
	private EntryBinding _dataBinding;
	private StoredClassCatalog _javaCatalog;
	
	private ArrayList<SecondaryDatabase> _indexes = new ArrayList<SecondaryDatabase>();
	private Database _catalogDb;
	private StoredMap _mapSec;
	
	public boolean add(OtpErlangTuple object) {
		final DatabaseEntry entry = new DatabaseEntry();
		_dataBinding.objectToEntry(object, entry);
		final byte[] data = entry.getData();

		try {
			final MessageDigest digest = MessageDigest.getInstance("MD5");
			digest.update(data);
			final byte[] hash = digest.digest();
			return _map.put(new OtpErlangBinary(hash), object)==null;
		} catch (NoSuchAlgorithmException e) {
			throw new TupleStoreException(e);
		}
	}
	public boolean addAll(Collection<? extends OtpErlangTuple> tuples) {
		boolean changed=false;
		for(OtpErlangTuple tuple: tuples)
		{
			changed|=add(tuple);
		}
		return changed;
	}
	public void clear() {
		tuples().clear();
	}
	public Comparator<? super OtpErlangTuple> comparator() {
		return tuples().comparator();
	}
	public boolean contains(Object arg0) {
		return tuples().contains(arg0);
	}
	public boolean containsAll(Collection<?> arg0) {
		return tuples().containsAll(arg0);
	}
	public boolean equals(Object arg0) {
		return tuples().equals(arg0);
	}
	public OtpErlangTuple first() {
		return tuples().first();
	}
	public int hashCode() {
		return tuples().hashCode();
	}
	public SortedSet<OtpErlangTuple> headSet(OtpErlangTuple toElement) {
		return tuples().headSet(toElement);
	}
	public boolean isEmpty() {
		return tuples().isEmpty();
	}
	public Iterator<OtpErlangTuple> iterator() {
		return tuples().iterator();
	}
	public OtpErlangTuple last() {
		return tuples().last();
	}
	public boolean remove(Object arg0) {
		return tuples().remove(arg0);
	}
	public boolean removeAll(Collection<?> arg0) {
		return tuples().removeAll(arg0);
	}
	public boolean retainAll(Collection<?> arg0) {
		return tuples().retainAll(arg0);
	}
	public int size() {
		return tuples().size();
	}
	public SortedSet<OtpErlangTuple> subSet(OtpErlangTuple fromElement,
			OtpErlangTuple toElement) {
		return tuples().subSet(fromElement, toElement);
	}
	public SortedSet<OtpErlangTuple> tailSet(OtpErlangTuple fromElement) {
		return tuples().tailSet(fromElement);
	}
	public Object[] toArray() {
		return tuples().toArray();
	}
	public <T> T[] toArray(T[] arg0) {
		return tuples().toArray(arg0);
	}
	/**
	 * @throws DatabaseException 
	 * @throws FileNotFoundException 
	 * 
	 */
	public BdbOtpTupleSet(Environment env, String databaseName) throws FileNotFoundException, DatabaseException {
		_env = env;

		//
		// Lets mimic the db.AccessExample 100%
		// and use plain old byte arrays to store the key and data strings.
		//

		//
		// Open a data store.
		//
		DatabaseConfig dbConfig = new DatabaseConfig();
		dbConfig.setAllowCreate(true);
		dbConfig.setType(DatabaseType.BTREE);
//		dbConfig.setHashNumElements(10000000);
//		int pagesize = 4096;
//		dbConfig.setPageSize(pagesize);
//		int average_key_size = 16;
//		int average_data_size = 54;
//		dbConfig.setHashFillFactor((pagesize - 32) / (average_key_size + average_data_size + 8));
//		dbConfig.setCacheSize(400000000);
		//dbConfig.setSortedDuplicates(true);
//		
//		dbConfig.setType(DatabaseType.BTREE);
		_catalogDb = _env.openDatabase(null, CLASS_CATALOG,null, dbConfig);
        _javaCatalog = new StoredClassCatalog(_catalogDb);
		
        _keyBinding = new SerialBinding(_javaCatalog, OtpErlangObject.class);
		_dataBinding = new SerialBinding(_javaCatalog, OtpErlangTuple.class);
//		_keyBinding = new OtpObjectBinding();
//		_dataBinding = new OtpObjectBinding();
        
		dbConfig.setType(DatabaseType.HASH);
        _db = _env.openDatabase(null, databaseName, null, dbConfig);
		
		
        


		
		//Secondary indices
		SecondaryConfig secConfig = new SecondaryConfig();
        secConfig.setAllowCreate(true);
        secConfig.setType(DatabaseType.HASH);
        secConfig.setSortedDuplicates(true);


        for(int i=0; i<4; i++)
        {
//        	if(i==1)
//        		continue;
	        secConfig.setKeyCreator(
	            new OtpErlangTupleSecondaryKeyCreator(_javaCatalog,i));
	        _indexes.add(_env.openSecondaryDatabase(
	        		null, 
	                INDEX_PREFIX+i,
	                null,
	                _db,
	                secConfig));
        }

		//
		// Now create a collection style map view of the data store
		// so that it is easy to work with the data in the database.
		//
		_map = new StoredMap(_db, _keyBinding, _dataBinding, true);
//		_mapSec = new StoredMap(_indexes.get(0), _keyBinding, _dataBinding, true);
	}
	@Override
	public int arity() {
		// TODO Auto-generated method stub
		return 0;
	}
	@Override
	public long removeBy(OtpErlangTuple fetchPattern) {
		// TODO Auto-generated method stub
		return 0;
	}
	@Override
	public Collection<OtpErlangTuple> fetch(OtpErlangTuple fetchPattern) {
		// TODO Auto-generated method stub
		return null;
	}
	@Override
	public Collection<OtpErlangTuple> select(OtpErlangTuple fetchPattern,
			OtpErlangTuple selectPattern) {
		// TODO Auto-generated method stub
		return null;
	}
	
	@SuppressWarnings("unchecked")
	private final SortedSet<OtpErlangTuple> tuples()
	{
		return (SortedSet<OtpErlangTuple>) _map.values();
	}

}
